package com.bkhech.bigdata.tiktok_anchor.videoinfo;

import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

import java.io.IOException;
import java.util.Arrays;
import java.util.Objects;

/**
 * 数据指标统计作业
 * 1：基于主播进行统计，统计每个主播在当天收到的总金币数量，总观看PV，总粉丝关注量，总视频开播时长;
 * <p>
 * 分析：
 * 1：为了方便统计主播的指标数据，最好是把这些字段整合到一个对象中，这样维护起来比较方便，这样就需要自定义 writable 了
 * 2：由于在这里需要一主播维度进行数据的聚合，所以需要以主播 ID 作为 key 进行聚合统计
 * 3：所以 Map 结点的 <k2, v2> 为 <Text, 自定义 Writable>
 * 4：由于需要聚合，所以 Reduce 阶段也需要有
 *
 * @author guowm
 * @date 2022/7/6
 */
@Slf4j
public class VideoInfoJob {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        //指定我们使用的文件系统类型:
        conf.set("fs.defaultFS", "hdfs://hadoop01:9000");
        FileSystem fileSystem = null;
        try {
            fileSystem = FileSystem.get(conf);
            run(args, fileSystem);
            fileSystem.close();
        } catch (IOException e) {
            e.printStackTrace();
            if (Objects.nonNull(fileSystem)) {
                try {
                    fileSystem.close();
                } catch (IOException ex) {
                    throw new RuntimeException(ex);
                }
            }
        }
    }

    public static void run(String[] args, FileSystem fileSystem) {
        // 可以指定job需要配置的参数
        final Configuration conf = fileSystem.getConf();
        try {
            final String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
            if (otherArgs.length < 2) {
                System.err.println("Usage: VideoInfoJob <in> [<in>...] <out>");
                System.exit(2);
            }

            // 参数
            String[] inputDirs = Arrays.copyOf(otherArgs, otherArgs.length - 1);
            String outputDir = otherArgs[otherArgs.length - 1];

            // 创建一个job
            final Job job = Job.getInstance(conf);
            // 注意：这一行必须设置，否则在集群中执行的是找不到 VideoInfoJob 这个类
            job.setJarByClass(VideoInfoJob.class);
            // 指定输入路径(可以是文件，也可以是目录)
            for (int i = 0; i < inputDirs.length; i++) {
                FileInputFormat.addInputPath(job, new Path(inputDirs[i]));
            }
            // 指定输出路径(只能指定一个不存在的目录)
            FileOutputFormat.setOutputPath(job, new Path(outputDir));

            // 指定map相关的代码
            job.setMapperClass(VideoInfoMap.class);
            // 指定k2的类型
            job.setMapOutputKeyClass(Text.class);
            // 指定v2的类型
            job.setMapOutputValueClass(VideoInfoWritable.class);

            // 指定 reduce 相关代码
            job.setReducerClass(VideoInfoReduce.class);
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(VideoInfoWritable.class);

            job.setNumReduceTasks(1);

            // 提交job
            final boolean waitForCompletion = job.waitForCompletion(true);
            log.info("---execute result: {}---", waitForCompletion);

        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
